﻿using System;
using Apache.NMS;
using log4net;
using Reactor.Environment;
using Reactor.Messaging.ServiceBus.Delivery;
using Reactor.Messaging.ServiceBus.Internals;
using Samurai.Wakizashi;

namespace Reactor.Messaging.ServiceBus.Receiving
{
    public abstract class ReceiverBase : DisposableBase, IReceiver
    {
        #region Fields

        protected static ILog Log;
        protected INmsGateway NmsGateway;
        protected ISession Session;
        protected IMessageConsumer Consumer;
        protected bool IsDisposed;

        #endregion

        /// <summary>
        /// Initializes a new instance of the <see cref="ReceiverBase"/> class.
        /// </summary>
        /// <param name="nmsGateway">The JMS manager responsible for managing the current JMS Connection and providing a valid session.</param>
        /// <param name="session">The JMS session from which to obtain message consumers, topics, queues, etc.</param>
        /// <param name="messagingChannel">The MessagingChannel.</param>
        protected ReceiverBase(INmsGateway nmsGateway, ISession session, IMessagingChannel messagingChannel)
        {
            if (nmsGateway == null) throw new ArgumentNullException("nmsGateway");
            if (session == null) throw new ArgumentNullException("session");
            if (messagingChannel == null) throw new ArgumentNullException("messagingChannel");

            Log = LogManager.GetLogger(GetType());
            NmsGateway = nmsGateway;
            Session = session;
            MessagingChannel = messagingChannel;
        }

        #region Implementation of IReceiver

        /// <summary>
        /// Occurs when a message is received.
        /// </summary>
        public event EventHandler<MessageEventArgs> MessageReceived;

        /// <summary>
        /// Starts this instance.
        /// </summary>
        public abstract void Start();

        /// <summary>
        /// Stops this instance.
        /// </summary>
        public virtual void Stop()
        {
            Dispose();
        }

        /// <summary>
        /// Gets a value indicating whether this instance is running.
        /// </summary>
        /// <value>
        /// 	<c>true</c> if this instance is running; otherwise, <c>false</c>.
        /// </value>
        public bool IsRunning { get; protected set; }

        /// <summary>
        /// Gets or sets the MessagingChannel.
        /// </summary>
        /// <value>The MessagingChannel.</value>
        public IMessagingChannel MessagingChannel { get; set; }

        #endregion

        #region Implementation of IDisposable

        /// <summary>
        /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
        /// </summary>
        /// <filterpriority>2</filterpriority>
        protected override void Dispose(bool disposing)
        {
            Consumer.Dispose();
            IsRunning = false;
            IsDisposed = true;

            base.Dispose(disposing);
        }

        #endregion

        protected virtual void OnReply(object reply, IMessageContext messageContext)
        {
            // Obtain original message so we can extract reply-to MessagingChannel and correlation id
            var textMessage = (ITextMessage)messageContext.RawMessage;
            var replyCorrelationId = textMessage.Properties.GetString("RequestCorrelationId");

            // If there is no reply-to MessagingChannel specified, we can't continue
            if (textMessage.NMSReplyTo == null || string.IsNullOrEmpty(replyCorrelationId))
            {
                if(Log.IsWarnEnabled) Log.WarnFormat("Cannot reply to message containing no reply-to MessagingChannel. Original message was sent to: {0}", textMessage.NMSDestination);
                return;
            }

            // Queue up reply transmission
            var transmissionTask = new TransmissionTask
                                       {
                                           MessagingChannel = Messaging.MessagingChannel.FromTypedDestinationName(textMessage.NMSReplyTo.ToString()),
                                           ReplyDestination = Context.ServiceBus.Configuration.CurrentEndpointInputChannel,
                                           Message = reply,
                                           RequestIdentifier = Guid.Parse(replyCorrelationId)
                                       };
            var taskDistributor = Context.ServiceLocator.GetInstance<TaskDistributor>();
            taskDistributor.EnqueueTask(transmissionTask);
        }

        protected virtual void OnMessage(IMessage message)
        {
            ITextMessage textMessage;
            try
            {
                if(Log.IsDebugEnabled) Log.DebugFormat("Received message destined for: {0}", message.NMSDestination);

                textMessage = (ITextMessage)message;
                Log.DebugFormat("Message details: {0}", textMessage.Text);

                // Extract actual message
                var msg = textMessage.ToObject();

                // Determine intended MessagingChannel
                var destination = Messaging.MessagingChannel.FromTypedDestinationName(message.NMSDestination.ToString());

                // Raise MessageReceived event if anyone is interested
                if (MessageReceived != null)
                    MessageReceived(this, new MessageEventArgs {Message = msg, MessagingChannel = destination});

                // Build up context
                var requestCorrelationId = textMessage.Properties.GetString("RequestCorrelationId");
                var msgContext = new MessageContext
                                     {
                                         Bus = Context.ServiceBus,
                                         RawMessage = message,
                                         OnReply = OnReply,
                                         RequestIdentifier = (!string.IsNullOrEmpty(requestCorrelationId)) ? Guid.Parse(requestCorrelationId) : Guid.Empty,
                                         NumberOfDeliveryAttempts = textMessage.GetNumberOfDeliveryAttempts(),
                                         MessagingChannel = destination,
                                         WorkflowInstanceId = (!string.IsNullOrEmpty(textMessage.NMSCorrelationID)) ? Guid.Parse(textMessage.NMSCorrelationID) : Guid.Empty
                };

                var deliveryCoordinator = new DeliveryCoordinator();
                deliveryCoordinator.DeliverMessage(msg, msgContext);
            }
            catch (Exception e)
            {
                Log.Error(string.Format("Unable to handle message from MessagingChannel: {0}", MessagingChannel.DestinationName), e);

                // If the intended MessagingChannel requires all message deliveries to be transactional, evaluate transactional status.
                if(MessagingChannel.IsTransactional)
                    HandleFailedTransaction(message);
            }
        }

        private void HandleFailedTransaction(IMessage message)
        {
            // Make sure we havent exceeded max delivery attempts
            var numberOfAttempts = message.GetNumberOfDeliveryAttempts();
            if (numberOfAttempts >= Context.ServiceBus.Configuration.MaxDeliveryAttempts)
                SendToDeadLetterQueue(message);
            else
                Resend(message, numberOfAttempts);
        }

        private void Resend(IMessage message, byte numberOfDeliveryAttemptsSoFar)
        {
            message.IncrementNumberOfDeliveryAttempts();

            // Create producer and resend
            var destination = Session.GetDestination(MessagingChannel.GetTypedDestinationName());
            using (var producer = Session.CreateProducer(destination))
            {
                producer.Send(message);
            }
            if (Log.IsDebugEnabled) Log.WarnFormat("Message of type: {0} resent to MessagingChannel: {1}. Number of delivery attempts so far: {2}", message.NMSType, message.NMSDestination, numberOfDeliveryAttemptsSoFar);
        }

        private void SendToDeadLetterQueue(IMessage message)
        {
            if(Context.ServiceBus.Configuration.DeadLetterChannel == null)
            {
                Log.ErrorFormat(string.Format("Unable to move message to dead letter queue because no DLQ was defined by the service bus. Dead message type: {0}", message.NMSType));
                return;
            }

            var destination = Session.GetDestination(Context.ServiceBus.Configuration.DeadLetterChannel.GetTypedDestinationName());
            using (var producer = Session.CreateProducer(destination))
            {
                producer.Send(message);
            }
            if (Log.IsDebugEnabled) Log.WarnFormat("Message of type: {0} sent to DLQ: {1} because the maximum number of delivery attempts was exceeded.", message.NMSType, message.NMSDestination);
        }
    }
}
